package com.wzy.gmail.realtime.app

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONObject}
import com.wzy.gmail.realtime.bean.{OrderDetail, OrderInfo, OrderWide}
import com.wzy.gmail.realtime.util.{MyEsUtils, MyKafkaUtils, MyOffsetUtils, MyRedisUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import java.time.{LocalDate, Period}
import java.util
import scala.collection.mutable.ListBuffer

/**
 * 订单宽表任务
 *
 * 1. 准备实时环境
 * 2. 从redis中读取offset * 2
 * 3.从kafka中消费数据 *2
 * 4.提取offset *2
 * 5.数据处理
 * 5.1 转换结构
 * 5.2 维度关联
 * 5.3 双流join
 * 6.写入es
 * 7.提交offset *2
 */
object DwdOrderApp {
  def main(args: Array[String]): Unit = {
    // 1. 准备环境
    val sparkConf: SparkConf = new SparkConf().setAppName("dwd_order_app").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    // 2. 读取offset
    // orderDetail
    val orderInfoTopicName: String = "DWD_ORDER_INFO_I_1018"
    val orderInfoGroup: String = "DWD_ORDER_INFO_GROUP"
    val orderInfoOffsets: Map[TopicPartition, Long] = MyOffsetUtils.readOffset(orderInfoTopicName, orderInfoGroup)
    // orderInfo
    val orderDetailTopicName: String = "DWD_ORDER_DETAIL_I_1018"
    val orderDetailGroup: String = "DWD_ORDER_DETAIL_GROUP"
    val orderDetailOffsets: Map[TopicPartition, Long] = MyOffsetUtils.readOffset(orderDetailTopicName, orderDetailGroup)

    // 3.从kafka中消费数据
    //order_info
    var orderInfoKafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if (orderInfoOffsets != null && orderDetailOffsets.nonEmpty) {
      orderInfoKafkaDStream = MyKafkaUtils.getKafkaDStream(orderInfoTopicName, ssc, orderInfoGroup, orderInfoOffsets)
    } else {
      orderInfoKafkaDStream = MyKafkaUtils.getKafkaDStream(orderInfoTopicName, ssc, orderInfoGroup)
    }

    //order_detail
    var orderDetailKafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if (orderDetailOffsets != null && orderDetailOffsets.nonEmpty) {
      orderDetailKafkaDStream = MyKafkaUtils.getKafkaDStream(orderDetailTopicName, ssc, orderInfoGroup, orderDetailOffsets)
    } else {
      orderDetailKafkaDStream = MyKafkaUtils.getKafkaDStream(orderDetailTopicName, ssc, orderDetailGroup)
    }

    //4.提取offset
    //order_info
    var orderInfoOffsetRanges: Array[OffsetRange] = null
    val orderInfoOffsetDStream: DStream[ConsumerRecord[String, String]] = orderInfoKafkaDStream.transform(
      rdd => {
        orderInfoOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )
    //order_detail
    var orderDetailOffsetRanges: Array[OffsetRange] = null
    val orderDetailOffsetDStream: DStream[ConsumerRecord[String, String]] = orderDetailKafkaDStream.transform(
      rdd => {
        orderDetailOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    // 5.处理数据
    // 5.1转换结构
    val orderInfoDStream: DStream[OrderInfo] = orderInfoOffsetDStream.map(
      consumerRecord => {
        val value: String = consumerRecord.value()
        val orderInfo: OrderInfo = JSON.parseObject(value, classOf[OrderInfo])
        orderInfo
      }
    )

    //    orderInfoDStream.print(100)

    val orderDetailDStream: DStream[OrderDetail] = orderDetailOffsetDStream.map(
      consumerRecord => {
        val value: String = consumerRecord.value()
        val orderDetail: OrderDetail = JSON.parseObject(value, classOf[OrderDetail])
        orderDetail
      }
    )
    //    orderDetailDStream.print(100)

    // 5.2 维度关联
    //orderInfo
    val orderInfoDimDStream: DStream[OrderInfo] = orderInfoDStream.mapPartitions(
      oderInfoIterator => {
        val jedis: Jedis = MyRedisUtils.getJedisFromPool()
        val orderInfoes: ListBuffer[OrderInfo] = ListBuffer[OrderInfo]()

        for (orderInfo <- oderInfoIterator) {
          // 关联用户维度
          val uid: Long = orderInfo.user_id
          val redisUserKey: String = s"DIM:USER_INFO:$uid"
          val userInfoJson: String = jedis.get(redisUserKey)
          val userInfoJsonObj: JSONObject = JSON.parseObject(userInfoJson)
          //提取性别
          val gender: String = userInfoJsonObj.getString("gender")
          //提取生日
          val birthday: String = userInfoJsonObj.getString("birthday")
          // 提取年龄
          val birthdayLd: LocalDate = LocalDate.parse(birthday)
          val nowLd: LocalDate = LocalDate.now()
          val period: Period = Period.between(birthdayLd, nowLd)
          val age: Int = period.getYears
          //补充到对象中
          orderInfo.user_gender = gender
          orderInfo.user_age = age

          //关联地区维度
          val provinceId: Long = orderInfo.province_id
          val redisProvinceKey: String = s"DIM:BASE_PROVINCE:$provinceId"
          val provinceJson: String = jedis.get(redisProvinceKey)
          val provinceJsonObj: JSONObject = JSON.parseObject(provinceJson)
          // 提取地区名称
          val provinceName: String = provinceJsonObj.getString("name")
          val provinceAreaCode: String = provinceJsonObj.getString("area_code")
          val province3166: String = provinceJsonObj.getString("iso_3166_2")
          val provinceIsoCode: String = provinceJsonObj.getString("iso_code")
          //补充到对象中
          orderInfo.province_name = provinceName
          orderInfo.province_area_code = provinceAreaCode
          orderInfo.province_3166_2_code = province3166
          orderInfo.province_iso_code = provinceIsoCode

          //处理日期字段
          val createTime: String = orderInfo.create_time
          val createDtHr: Array[String] = createTime.split(" ")
          val createDate: String = createDtHr(0)
          val createHr: String = createDtHr(1).split(":")(0)
          // 补充到对象中
          orderInfo.create_date = createDate
          orderInfo.create_hour = createHr
          orderInfoes.append(orderInfo)
        }
        jedis.close()
        orderInfoes.iterator
      }
    )
    //    orderInfoDimDStream.print(100)

    // 5.3双流join
    // 内连接
    // 外连接
    //  左外联
    //  右外联
    //  全外联
    // 从数据库层面orderInfo 和orderDetail一定可以关联成功
    // 从流处理层面 orderInfo 和orderDetail是两个流，只有在一个批次才可以join成功
    // 不是一个批次就会join不成功
    // 数据延迟导致的数据没有进入一个批次，在实时处理中是正常现象，可以接收因为延迟导致的最终结果延迟
    // 不能接收因为延迟导致的数据丢失

    val oderInfoKVDStream: DStream[(Long, OrderInfo)] = orderInfoDimDStream.map(orderInfo => (orderInfo.id, orderInfo))
    val orderDetailKVDSteam: DStream[(Long, OrderDetail)] = orderDetailDStream.map(orderDetail => (orderDetail.order_id, orderDetail))
    //    val orderJoinDStream: DStream[(Long, (OrderInfo, OrderDetail))] = oderInfoKVDStream.join(orderDetailKVDSteam)
    //    orderJoinDStream.print(100)
    // 解决：首先使用fullOuterJoin，保证join成功或者没有成功的数据出现到结果中
    // 1. 扩大采集周期，治标不治本
    // 2. 使用窗口，治标不治本，考虑数据去重，spark状态的缺点
    // 3. 首先使用fullOuterJoin，保证join成功或者没有成功的数据都出现到结果中
    //    让双方都多两部操作，到缓存中找
    val orderJoinDStream: DStream[(Long, (Option[OrderInfo], Option[OrderDetail]))] = oderInfoKVDStream.fullOuterJoin(orderDetailKVDSteam)

    val orderWideDStream: DStream[OrderWide] = orderJoinDStream.mapPartitions(
      orderJoinIter => {
        val orderWides: ListBuffer[OrderWide] = ListBuffer[OrderWide]()
        val jedis: Jedis = MyRedisUtils.getJedisFromPool()
        for ((key, (orderInfoOp, orderDetailOp)) <- orderJoinIter) {
          //OrderInfo有OrderDetail有
          if (orderInfoOp.isDefined) {
            //取出orderInfo
            val orderInfo: OrderInfo = orderInfoOp.get
            if (orderDetailOp.isDefined) {
              //取出orderDetail
              val orderDetail: OrderDetail = orderDetailOp.get
              //组装成orderWide
              val orderWide = new OrderWide(orderInfo, orderDetail)
              orderWides.append(orderWide)
            }
            //OrderInfo有OrderDetail没有

            //orderInfo写缓存
            // 类型：String
            // key:ORDERJOIN:ORDER_INFO:ID
            //value:json
            // 写入api:set
            // 读取api：get
            // 是否过期：24h
            val redisOrderInfoKey: String = s"ORDERJOIN:ORDER_INFO:${orderInfo.id}"
            jedis.setex(redisOrderInfoKey, 24 * 3600, JSON.toJSONString(orderInfo, new SerializeConfig(true)))

            // orderInfo读缓存
            val redisOrderDetailKey: String = s"ORDERJOIN_ORDER_RETAIL:${orderInfo.id}"
            val orderDetails: util.Set[String] = jedis.smembers(redisOrderDetailKey)
            if (orderDetails != null && orderDetails.size() > 0) {
              import scala.collection.JavaConverters._
              for (orderDetailJson <- orderDetails.asScala) {
                val orderDetail: OrderDetail = JSON.parseObject(orderDetailJson, classOf[OrderDetail])
                // 组装成orderWide
                val orderWide = new OrderWide(orderInfo, orderDetail)
                // 加入到结果集中
                orderWides.append(orderWide)
              }
            }

          } else {
            // orderInfo有，orderDetail有
            val orderDetail: OrderDetail = orderDetailOp.get
            // 读缓存
            val redisOrderInfoKey: String = s"ORDERJOIN:ORDER_INFO:${orderDetail.id}"
            val orderInfoJson: String = jedis.get(redisOrderInfoKey)
            if (orderInfoJson != null && orderInfoJson.size > 0) {
              val orderInfo: OrderInfo = JSON.parseObject(orderInfoJson, classOf[OrderInfo])
              // 组装成orderWide
              val orderWide = new OrderWide(orderInfo, orderDetail)
              orderWides.append(orderWide)
            } else {
              // OrderInfo在redis中没有OrderDetail有
              // 写缓存
              // 类型：set
              // key:ORDERJOIN_ORDER_RETAIL:ORDER_ID
              // value:json,json...
              // 写入：sadd
              // 读取：smembers
              // 过期：24h
              val redisOrderDetailKey: String = s"ORDERJOIN_ORDER_RETAIL:${orderDetail.order_id}"
              jedis.sadd(redisOrderDetailKey, JSON.toJSONString(orderDetail, new SerializeConfig(true)))
              jedis.expire(redisOrderDetailKey, 24 * 3600)
            }
          }
        }
        jedis.close()
        orderWides.iterator
      }
    )
    //    orderWideDStream.print(1000)
    // 6.写入ES
    // 1. 索引分割，通过索引模板控制mapping setting aliases
    // 2. 使用工具类将数据写入es
    orderWideDStream.foreachRDD(
      rdd => {
        rdd.foreachPartition(
          orderWideIter => {
            val orderWides: List[(String, OrderWide)] = orderWideIter.map(orderWide => (orderWide.detail_id.toString, orderWide)).toList
            if (orderWides.nonEmpty) {
              val head: (String, OrderWide) = orderWides.head
              val data: String = head._2.create_date
              // 索引名
              val indexName: String = s"gmall_order_wide_1018_$data"
              //写入es
              MyEsUtils.bulkSave(indexName, orderWides)
            }
          }
        )
        //提交offsets
        MyOffsetUtils.saveOffset(orderInfoTopicName, orderInfoGroup, orderInfoOffsetRanges)
        MyOffsetUtils.saveOffset(orderDetailTopicName, orderDetailGroup, orderDetailOffsetRanges)
      }
    )

    ssc.start()
    ssc.awaitTermination()
  }

}
